/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.analysys.presto.connector.hbase;

import javax.inject.Inject;

import com.analysys.presto.connector.hbase.pool.SimpleDataSource;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import io.airlift.log.Logger;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.ClusterMetrics;
import org.apache.hadoop.hbase.ClusterStatus;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * HBase client manager
 *
 * @author wupeng
 * @date 2019/01/29
 */
public class HBaseClientManager {

    private static final Logger log = Logger.get(HBaseClientManager.class);

    /**
     * hbase Connector conf
     */
    private final HBaseConfig config;

    /**
     * HBase 连接池
     */
    private final SimpleDataSource dataSource;

    /**
     * hbase server
     */
    private final List<ServerName> serverNames = new ArrayList<>();


    /**
     * Hbase table cache
     */
    private final Cache<String, HBaseTable> TABLE_CACHED = CacheBuilder.newBuilder()
            .initialCapacity(2000)
            .maximumSize(2000)
            .expireAfterAccess(5, TimeUnit.MINUTES)
            .build();

    @Inject
    public HBaseClientManager(HBaseConfig config) {
        this.config = config;
        final Properties poolProperties = new Properties();
        poolProperties.setProperty("pool.max", ""+config.getConnectionPoolMax());
        poolProperties.setProperty("pool.min", ""+config.getConnectionPoolMin());
        this.dataSource = new SimpleDataSource(this, poolProperties);
    }


    /**
     * 从DS 获取一个可用的连接
     * @return
     */
    public Connection createConnection() {
        return dataSource.getConnection();
    }

    /**
     * 创建一个新的链接
     * @return
     */
    public Connection getOrCreateConnection() throws IOException {
        return createConnectionForce(new AtomicInteger(0));
    }


    Connection createConnectionForce(AtomicInteger deep) throws IOException {
        Configuration conf;
        Connection connection;
        try {
            conf = HBaseConfiguration.create();

            final String hbaseConfigResources = config.getHbaseConfigResources();
            if(StringUtils.isNotBlank(hbaseConfigResources)) {
                final String[] resources = hbaseConfigResources.split(",");
                for (String resource : resources) {
                    String resourceName = StringUtils.trimToNull(resource);
                    if(StringUtils.isNotBlank(resourceName)) {
                        File resourceFile = new File(resourceName);
                        if(resourceFile.exists()) {
                            conf.addResource(resourceFile.toURL());
                            log.info("add hbase conf resource: " + resource);
                        }
                    }
                }

                conf.reloadConfiguration();
            }

            // 创建 HBase 连接
            long startTime = System.currentTimeMillis();
            connection = ConnectionFactory.createConnection(conf);
            log.info("Create HBase connection " + (connection == null ? "failed." : "succeed.")
                    + ", used " + (System.currentTimeMillis() - startTime) + " mill sec");

            serverNames.clear();
            // 获取所有 Region server
            try(Admin admin = connection.getAdmin();) {
                final ClusterMetrics clusterStatus = admin.getClusterMetrics();
                serverNames.addAll(clusterStatus.getLiveServerMetrics().keySet());
                for (ServerName serverName : serverNames) {
                    log.info("list hbase server: " + serverName.getAddress());
                }
            }
            return connection;
        } catch (Exception ex) {
            if(ex instanceof IOException) {
                throw (IOException)ex;
            } else {
                throw new IOException(ex);
            }
        }
    }


    /**
     * 获取
     * @return
     */
    public Admin getAdmin() {
        return getAdmin(new AtomicInteger(0));
    }

    Admin getAdmin(AtomicInteger deep) {
        deep.incrementAndGet();
        Admin admin = null;
        try(Connection connection = createConnection()) {
            admin = connection.getAdmin();
            admin.listNamespaceDescriptors();
            return admin;
        } catch (Exception ex) {
            log.error(ex, ex.getMessage());
            if(deep.get() >= 2) {
                return null;
            }
            admin = getAdmin(deep);
        }
        return admin;
    }

    public HBaseTable getTable(String schema, String tableName) {
        Objects.requireNonNull(schema, "schema is null");
        Objects.requireNonNull(tableName, "tableName is null");
        TableName htableName = TableName.valueOf(schema.getBytes(), tableName.getBytes());
        String cacheKey = htableName.getNamespaceAsString() + ":" + htableName.getNameAsString();
        HBaseTable table = TABLE_CACHED.getIfPresent(cacheKey);

        // 检测 HBase 是否存在改表
        HTableDescriptor hTableDescriptor = null;
        Admin admin = null;
        try {
            admin = this.getAdmin();
            hTableDescriptor = admin.getTableDescriptor(htableName);
        } catch (IOException ex) {
            log.error(ex, ex.getMessage());
        } finally {
            if (admin != null) {
                this.close(admin);
            }
        }

        if (hTableDescriptor == null) {
            return null;
        }

        // 检查 meta 是否有更新
        if(table != null) {
            try {
                // 缓存更新
                HBaseTable newTable = new HBaseTable(schema, hTableDescriptor, config);
                if (newTable.getLastModify() > table.getLastModify()) {
                    // 有更新 meta
                    table = newTable;
                    TABLE_CACHED.put(cacheKey, table);
                    log.warn("hbase table: " + schema + ":" + tableName + " meta info changed, reload meta info!");
                }
            } catch (Exception e) {
                log.warn("can not found hbase table: " + schema + ":" + tableName, e);
            }
            return table;
        } else {
            // 第一次加载
            table = new HBaseTable(schema, hTableDescriptor, config);
            TABLE_CACHED.put(cacheKey, table);
            return table;
        }
    }

    void close(Admin admin) {
        try {
            admin.close();
        } catch (Exception ex) {
            log.error(ex, ex.getMessage());
        }
    }

    /**
     * close all hbase connection
     */
    public void close() {
        try {
            // close all hbase connection
            dataSource.close();
        } catch (Exception ex) {
            log.error(ex, ex.getMessage());
        }
    }

    public String getServerRandom() {
        final ServerName serverName = serverNames.get(new Random().nextInt(serverNames.size()));
        return serverName.getHostAndPort();
    }

    public HBaseConfig getConfig() {
        return config;
    }
}
